package com.spark.mysql

import java.sql.DriverManager
import java.util.Properties

import org.apache.hadoop.hbase.client.{ConnectionFactory, Result, Scan}
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}

object insertfunction {

  /** 连接数据库
    *
    */
  val url = "jdbc:mysql://localhost:3306/test?rewriteBatchedStatements=true&" +
    "useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull" +
    "&useSSL=false"
  val driver = "com.mysql.jdbc.Driver"
  val user = "root"
  val password = "root"

  val prop = new Properties
  prop.setProperty("user", user)
  prop.setProperty("password", password)
  prop.setProperty("driver", driver)
  prop.setProperty("charset", "UTF-8")
  val conn = DriverManager.getConnection(url, prop)

  /**
    * hbase数据库连接
    */
  val c = HBaseConfiguration.create()
  val connection = ConnectionFactory.createConnection(c)
  val tableName = TableName.valueOf("fwwb")

  /**
    * 返回一个迭代器,通过迭代器可获取一个时间段的数据
    *
    * @param start 开始时间戳
    * @param end   结束时间戳
    * @return result 迭代器
    */
  def selectByTime(start: Long, end: Long): java.util.Iterator[Result] = {
    val table = connection.getTable(tableName)
    val scan = new Scan()
    scan.setTimeRange(start, end)
    val scanner = table.getScanner(scan)
    val result = scanner.iterator()
    result
  }

  /**
    * 使用PreparedStatement将数据插入tmpcount
    *
    * @param p
    */
  def insertToTmp(p: (String, String, String)) {
    try {
      val sql = "insert into tmpcount(time,ll,imsi)values(?,?,?)"
      var ps = conn.prepareStatement(sql)
      //设置占位符对应的值
      ps.setString(1, p._1)
      ps.setString(2, p._2)
      ps.setString(3, p._3)
      ps.executeUpdate()
      ps.close()
    } catch {
      case e: Exception => e.printStackTrace()
        conn.rollback()
    }
  }

  /**
    * 使用PreparedStatement将数据插入count
    *
    * @param p
    */
  def insertToCount(p: (String, Float, Float, Int)) {
    try {
      val sql = "insert into current_count(time,longitude,latitude,count)values(?,?,?,?)"
      var ps = conn.prepareStatement(sql)
      //设置占位符对应的值
      ps.setString(1, p._1)
      ps.setFloat(2, p._2)
      ps.setFloat(3, p._3)
      ps.setInt(4, p._4)
      println(ps)
      ps.executeUpdate()
      ps.close()
    } catch {
      case e: Exception => e.printStackTrace()
        conn.rollback()
    }
  }

  /**
    * 使用PreparedStatement将数据插入historycount
    *
    * @param p
    */
  def insertToHiscount(p: (String, Float, Float, Int)) {
    try {
      val sql = "insert into history_count(time,longitude,latitude,count)values(?,?,?,?) ON DUPLICATE KEY UPDATE count = ?" // 错误， 原为longtitude
      var ps = conn.prepareStatement(sql)
      //设置占位符对应的值
      ps.setString(1, p._1)
      ps.setFloat(2, p._2)
      ps.setFloat(3, p._3)
      ps.setInt(4, p._4)
      ps.setInt(5, p._4)
      ps.executeUpdate()
      ps.close()
    } catch {
      case e: Exception => e.printStackTrace()
        conn.rollback()
    }
  }

  /**
    * 将数据插入zhuliu
    *
    * @param p
    */
  def insertTozhuliu(p: (String, String, String)) {
    try {
      val sql = "insert into zhuliudata(time,ll,imsi)values(?,?,?)"
      var ps = conn.prepareStatement(sql)
      //设置占位符对应的值
      ps.setString(1, p._1)
      ps.setString(2, p._2)
      ps.setString(3, p._3)
      ps.executeUpdate()
      ps.close()
    } catch {
      case e: Exception => e.printStackTrace()
        conn.rollback()
    }
  }

  /**
    * 将数据插入Staytime
    *
    * @param p
    */
  def insertToStaytime(p: (String, String, String, String, Long, Int)) {
    try {
      val sql = "insert into staytime(startTime,endTime,ll,imsi,staytime,type)values(?,?,?,?,?,?)"
      var ps = conn.prepareStatement(sql)
      //设置占位符对应的值
      ps.setString(1, p._1)
      ps.setString(2, p._2)
      ps.setString(3, p._3)
      ps.setString(4, p._4)
      ps.setLong(5, p._5)
      ps.setInt(6, p._6)
      ps.executeUpdate()
      ps.close()
    } catch {
      case e: Exception => e.printStackTrace()
        conn.rollback()
    }
  }

  /**
    * 将数据插入analyse
    *
    * @param p
    */
  def insertToAnalyse(p: (String, Long , Int)) {
    try {
      val sql = "insert into analyse (ll,avgtime,type) values(?,?,?)"
      var ps = conn.prepareStatement(sql)
      //设置占位符对应的值
      ps.setString(1, p._1)
      ps.setLong(2, p._2)
      ps.setInt(3, p._3)
      ps.executeUpdate()
      ps.close()
    } catch {
      case e: Exception => e.printStackTrace()
        conn.rollback()
    }
  }

  /**
    * 将数据插入Final
    *
    * @param p
    */
  def insertToFinal(p: (String, Long, Long)) {
    try {
      val sql = "insert into final(ll,daytime,nighttime)values(?,?,?)"
      var ps = conn.prepareStatement(sql)
      //设置占位符对应的值
      ps.setString(1, p._1)
      ps.setLong(2, p._2)
      ps.setLong(3, p._3)
      ps.executeUpdate()
      ps.close()
    } catch {
      case e: Exception => e.printStackTrace()
        conn.rollback()
    }
  }

  def insertToAbnormal(p: (String, Double, Double, Float, Float)): Unit = {
    try {
      val sql = "insert into abnormal(time,longitude,latitude,count,rate)values(?,?,?,?,?)"
      var ps = conn.prepareStatement(sql)
      //设置占位符对应的值
      ps.setString(1, p._1)
      ps.setDouble(2, p._2)
      ps.setDouble(3, p._3)
      ps.setInt(4, p._4.toInt)
      ps.setInt(5, p._5.toInt)
      println(ps)
      ps.executeUpdate()
      ps.close()
    } catch {
      case e: Exception => e.printStackTrace()
        conn.rollback()
    }
  }

  def insertToCompare(nextTime: String, longitude: String, latitude: String, finalNum: Int): Unit = {
    try {
      val sql = "insert into compare(time, longitude, latitude, average) values (?,?,?,?) ON DUPLICATE KEY UPDATE average = ?"
      val ps = conn.prepareStatement(sql)
      ps.setString(1, nextTime)
      ps.setDouble(2, longitude.toDouble)
      ps.setDouble(3, latitude.toDouble)
      ps.setInt(4, finalNum)
      ps.setInt(5, finalNum)
      ps.executeUpdate()

      printf("insert into compare(time, longitude, latitude, average) values (%s,%s,%s,%s)  ON DUPLICATE KEY UPDATE average = %s", nextTime, longitude, latitude, finalNum, finalNum)
      println()
      println()
    } catch {
      case e: Exception => e.printStackTrace()
        conn.rollback()
    }
  }

  
  /**
    * 将数据插入analyse
    *
    * @param p
    */
  def insertToNewanalyse(p: (String, Long , Int)) {
    try {
      val sql = "insert into newanalyse (ll,time,type) values(?,?,?)"
      var ps = conn.prepareStatement(sql)
      //设置占位符对应的值
      ps.setString(1, p._1)
      ps.setLong(2, p._2)
      ps.setInt(3, p._3)
      ps.executeUpdate()
      ps.close()
    } catch {
      case e: Exception => e.printStackTrace()
        conn.rollback()
    }
  }
}